package com.spzx.product.listener;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.rabbitmq.client.Channel;
import com.spzx.common.rabbitmq.service.RabbitmqService;
import com.spzx.order.domain.OrderInfo;
import com.spzx.product.domain.dto.SkuStockDTO;
import com.spzx.product.mapper.SkuStockMapper;
import jakarta.annotation.Resource;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
public class SpzxProductMqListener {
    @Resource
    SkuStockMapper skuStockMapper;
    @Resource
    RedisTemplate redisTemplate;

    /*
                    rabbitmqService.sendMsg("spzx.payment.exchange",
                        "spzx.payment.pay.success" , map);
     */
    //支付成功 更新库存和销量
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "spzx.product.pay.success.queue",durable = "true"),
                    exchange = @Exchange(value = "spzx.payment.exchange" ,
                            type = ExchangeTypes.TOPIC),//交换机
                    key = "spzx.payment.pay.success"
            )

    }) //一旦队列中有消息  消费者会被回调
    public void orderPaySuccessConsumer(Channel channel , Map<String,Object> map , Message message) throws IOException {
        String orderNo = map.get("orderNo").toString();
        try {
            //1、使用消息: 更新订单的状态
            //rabbitmq幂等性如何保证： 使用消息中唯一的值 结合redis的setnx来实现
            Boolean flag = redisTemplate.opsForValue().setIfAbsent("spzx:product:pay:" + orderNo, "1",
                    10, TimeUnit.MINUTES);
            if(flag){
                //第一次执行业务
                //更新库存和销量
                //"spzx:order:stock:cache:"+skuStockDTOS.get(0).getOrderNo()
                Object o = redisTemplate.opsForValue().get("spzx:order:stock:cache:" + orderNo);
                if(!ObjectUtils.isEmpty(o)){
                    List<SkuStockDTO> skuStockDTOS = (List<SkuStockDTO>) o;
                    //更新库存和销量
                    skuStockMapper.updateStockAndSales(skuStockDTOS);
                }
            }


            //2、手动ack： 告诉mq消息已经使用完毕 可以删除
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),
                    false);//prefetch一次可以获取多个消息， 参数2 配置是否要等这一批的多个消息一起提交ack
        } catch (Exception e) {
//            throw new RuntimeException(e);
            //如果出现异常，可以让消息归队 再次尝试消费 如果多次消费失败可以丢弃消息
            if(message.getMessageProperties().isRedelivered()){
                //消息是重新进入队列的，丢弃消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            }else{
                //删除幂等性分布式锁： 消息重新被消费
                redisTemplate.delete("spzx:product:pay:" + orderNo);
                //消息第一次消费出现异常，重新归队 再次消费消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                        false , true);//参数3：重新进入队列
            }

        }


    }



    //延迟关单消费者
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "spzx.product.stock.release.queue"),
                    exchange = @Exchange(value = "spzx.product.exchange",
                            type = ExchangeTypes.TOPIC),
                    key = "spzx.order.close.success"
            )
    })
    public void closeOrderConsumer(Channel channel , Map<String,Object> map,
                                   Message message) throws IOException {
        String orderNo = map.get("orderNo").toString();
        try {
            //1、释放库存的业务
            //幂等性校验
            Boolean flag = redisTemplate.opsForValue().setIfAbsent("spzx:order:stock:release:lock:" + orderNo,
                    "1", 10, TimeUnit.MINUTES);
            if(!flag){
                return;
                //rabbitmq如何保证幂等性：消费者中通过消息中唯一的值 结合redis setnx实现
                // 生产者重试可能会导致幂等性问题
                // 消费者重试： 消费异常 再次消费(需要再catch中删除幂等性的token)
            }
            //1.1 获取redis中缓存的库存
            Object o = redisTemplate.opsForValue().get("spzx:order:stock:cache:" + orderNo);
            if(o==null){
                //读取库存的缓存失败：因为支付等操作/多次提交释放库存的消息 导致重复进入当前的消费者业务
                return;
            }
            List<SkuStockDTO> skuStockDTOS = (List<SkuStockDTO>) o;
            //1.2 执行释放库存的业务
            skuStockDTOS.forEach(skuStockDTO -> {
                skuStockMapper.releaseStock(skuStockDTO);
            });
            //1.3 删除释放过的库存缓存
            redisTemplate.delete("spzx:order:stock:cache:" + orderNo);
            //2、手动ack： 告诉mq消息已经使用完毕 可以删除
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),
                    false);//prefetch一次可以获取多个消息， 参数2 配置是否要等这一批的多个消息一起提交ack
        } catch (Exception e) {
//            throw new RuntimeException(e);
            //如果出现异常，可以让消息归队 再次尝试消费 如果多次消费失败可以丢弃消息
            if(message.getMessageProperties().isRedelivered()){
                //消息是重新进入队列的，丢弃消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            }else{
                //删除幂等性的token 避免消费者不能再次消费
                redisTemplate.delete("spzx:order:stock:release:lock:" + orderNo);
                //消息第一次消费出现异常，重新归队 再次消费消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                        false , true);//参数3：重新进入队列

            }

        }

    }
}
